package com.dky.satell.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.dky.satell.common.KafkaKeyUtil;
import com.dky.satell.entity.MessageConfigMeta;
import com.dky.satell.entity.OfflineFileMeta;
import com.dky.satell.kafka.model.MessageWrapper;
import com.dky.satell.mapper.LocalTaskMapper;
import com.dky.satell.service.MessageConfigMetaService;
import com.dky.satell.entity.MessageConfig;
import com.dky.satell.kafka.model.OfflineFileWarpper;
import com.dky.satell.service.OfflineFileMetaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * kafka 消费者
 *
 * @author lihaobin
 * @version 1.0
 * @date 2021/4/20 10:17
 */
@Slf4j
@Component
public class KafkaConsumer {

    @Value("${kafka.store.topicPrefix}")
    private String storeTopicPrefix;

    @Resource
    private KafkaTemplate kafkaTemplate;

    @Resource
    private OfflineFileMetaService offlineFileMetaService;

    @Resource
    private MessageConfigMetaService messageConfigMetaService;

    @Resource
    private LocalTaskMapper localTaskMapper;

    @KafkaListener(topics = {"${kafka.transfer.topic.file}"})
    public void fileListener(ConsumerRecord<String, String> record) {
        log.info("接收kafka消息：topic = [{}], offset = [{}], value = [{}]", record.topic(), record.offset(), record.value());
        // 从中获取消息
        String srcZoneId = KafkaKeyUtil.zoneByKey(record.key());
        String value = record.value();
        OfflineFileWarpper offlineFileWarpper = JSON.parseObject(value, OfflineFileWarpper.class);

        // 元数据入库
        OfflineFileMeta offlineFileMeta = new OfflineFileMeta();
        BeanUtils.copyProperties(offlineFileWarpper.getOfflineFile(), offlineFileMeta);
        offlineFileMeta.setCreateTime(null);
        offlineFileMeta.setUpdateTime(null);
        offlineFileMeta.setRegion(srcZoneId);
        offlineFileMetaService.saveOrUpdate(offlineFileMeta);

//        String proxyFileName = offlineFileWarpper.getOfflineFile().getProxyFileName();
//        LocalTask localTask = new LocalTask();
//        localTask.setType(FileDownloader.FILE_DOWNLOAD_TYPE);
//        localTask.setContent(proxyFileName);
//        localTaskMapper.insert(localTask);
    }

    /**
     * 接收采集端消息配置数据
     */
    @KafkaListener(topics = {"${kafka.transfer.topic.messageConfig}"})
    public void messageConfigListener(ConsumerRecord<String, String> record) {
        log.info("接收kafka消息： topic = {}, offset = {}, value = {}", record.topic(), record.offset(), record.value());
        // 从中获取消息
        String srcZoneId = KafkaKeyUtil.zoneByKey(record.key());
        String value = record.value();
        MessageConfig messageConfig = JSON.parseObject(value, MessageConfig.class);

        // 元数据入库
        MessageConfigMeta messageConfigMeta = new MessageConfigMeta();
        BeanUtils.copyProperties(messageConfig, messageConfigMeta);
        messageConfigMeta.setCreateTime(null);
        messageConfigMeta.setUpdateTime(null);
        messageConfigMeta.setRegion(srcZoneId);
        messageConfigMetaService.saveOrUpdate(messageConfigMeta);
    }

    /**
     * 接收采集端数据，写入数据需求方消息队列
     */
    @KafkaListener(topics = {"${kafka.transfer.topic.message}"})
    public void messageListener(ConsumerRecord<String, String> record) {
        log.info("接收kafka消息：topic = {}, offset = {}, value = {}", record.topic(), record.offset(), record.value());
        String value = record.value();
        MessageWrapper messageWrapper = JSON.parseObject(value, MessageWrapper.class);

        // 推送至使用方消息队列
        String storeTopic = storeTopicPrefix + messageWrapper.getMessageConfig().getTopic();
        kafkaTemplate.send(storeTopic, messageWrapper.getPayload());
    }

}
